/**
 * Project Name:alqsoft-ms
 * File Name:ServerConnection.java
 * Package Name:com.alqsoft
 * Date:2016年10月18日下午2:48:42
 * Copyright (c) 2016, chenzhou1025@126.com All Rights Reserved.
 *
*/

package com.alqsoft;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import com.alibaba.fastjson.JSON;
import com.alqsoft.bean.BaseInfo;
import com.alqsoft.interceptor.BaseInfoInterface;
import com.alqsoft.interceptor.BaseInfoProxyFactory;
import com.alqsoft.redis.RedisUtils;
import com.alqsoft.util.BaseBeanUtils;

import redis.clients.jedis.Jedis;

/**
 * ClassName:ServerConnection <br/>
 * Function: TODO ADD FUNCTION. <br/>
 * Reason:	 TODO ADD REASON. <br/>
 * Date:     2016年10月18日 下午2:48:42 <br/>
 * @author   张灿
 * @version  打开server通道，针对spring 配置bean 初始化参数
 * @since    JDK 1.8
 * @see 	 
 */
public class ServerConnection implements ConnectionInterface{
	
	private Charset charset = Charset.forName("UTF-8");
	
	private ServerProperties serverProperties;
	
	private Selector selector = null;
	
	ConcurrentHashMap<String,Object> hashMap = new ConcurrentHashMap<String,Object>();//保存有效的SocketChannel
	
	private AtomicInteger i=new AtomicInteger(0);
	
	/**
	 * 
	 * @param serverProperties
	 */
	public ServerConnection(ServerProperties serverProperties){
		this.serverProperties = serverProperties;
	}
	/**
	 * 打开nio服务器，监听端口，发送和接收消息
	 */
	public void open(){
		new ThreadResponseSend();
		new ThreadFilterSocketChannel();
		new ThreadServerBuffer();
		this.init();
	}
	/**
	 * 初始化监听端口
	 */
	public void init(){
		try {
			selector = Selector.open();//打开监听器,监听各读 写事件；
			ServerSocketChannel server = ServerSocketChannel.open();//打开通道
			System.out.println("服务器ip="+serverProperties.getServerIp()+":"+serverProperties.getServerPort());
			InetSocketAddress isa = new InetSocketAddress(serverProperties.getServerIp(), serverProperties.getServerPort());//监听服务器端口
			server.socket().bind(isa);//服务器端口和通道进行绑定
			server.configureBlocking(false);//设置为非阻塞模式
	        // 将server注册到指定Selector对象  
	        server.register(selector, SelectionKey.OP_ACCEPT);
	        while(selector.select() > 0){
	        	Set<SelectionKey> keys=selector.selectedKeys();  
	        	Iterator<SelectionKey> it = keys.iterator();  
	        	while(it.hasNext()) {
	        		SelectionKey sk=(SelectionKey) it.next();
	        		it.remove();
	        		try{
	        			if(sk.isAcceptable()){
							System.out.println("打开了连接通道");
							SocketChannel sc = server.accept();
							sc.configureBlocking(false);
							sc.register(selector, SelectionKey.OP_READ);
							sk.interestOps(SelectionKey.OP_ACCEPT);
							i.addAndGet(1);
						}
						if(sk.isReadable()){
							SocketChannel sc = (SocketChannel) sk.channel();
							ByteBuffer intBuff = ByteBuffer.allocate(4);    //读取INT头信息的缓存池 
							ByteBuffer buff = ByteBuffer.allocate(1024);     //读取OBJ有效数据的缓存池
						    //有效数据长度  
						    int ObjLength = 0;  
						    //从NIO信道中读出的数据长度  
						    int readObj = 0; 
							String content = "";
			        	   while(sc.read(intBuff)!=0){
//					        		   获取报文真正长度
			        		 ObjLength = intBuff.getInt(0);
			        		 buff.limit(ObjLength);
			        		 while (sc.read(buff)!=0) {
			        			 buff.flip();
			        			 content+=charset.decode(buff);
		                     }  
		                     System.out.println("收到客户端消息"+content);
						     // 如果content的长度大于0，即聊天信息不为空  
					           if (content.length() > 0) {
					        	   try{
					        		   BaseInfoInterface baseInfoInterface = BaseInfoProxyFactory.getBaseInfoInterface(serverProperties.loadInterceptor());
						        	   BaseInfo baseInfo = baseInfoInterface.doFilter(content);
//					                            进行业务判断 0登录 1接受客户端发送内容 2心跳包 3注销 -1错误提示，消息异常，加密异常等错误信息
			                            switch(baseInfo.getEvent()){
			                            
			                            case -1:
			                            	System.out.println("消息异常");
			                            case 0:
			                            	responseLogin(sc,baseInfo,"123456");
			                            	break;
			                            case 2:
			                            	break;
			                            case 1:	
//			                            	后续需要改成异步的，支持分布式；这个地方瓶颈很容易达到；
			                            	receiveSend(baseInfo);
			                            }
					        	   }catch(Exception e){
					        		   e.printStackTrace();
					        		   System.out.println(content);
					        		   System.out.println("数据接收异常！！！！！！！");
					        	   }
			                    }
			        	   }
	                        // 将sk对应的Channel设置成准备下一次读取  
	                        sk.interestOps(SelectionKey.OP_READ);  
						}
	        		}catch(Exception e){
	        			e.printStackTrace();
                    	System.out.println(e.getMessage());
                        sk.cancel();  
                        if (sk.channel() != null) {  
                            sk.channel().close();  
                        }
	        			e.printStackTrace();
	        		}finally{
	        			selector.selectedKeys().remove(sk);
	        		}
					
				}
	        }
			
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println("打开选择器失败！！！");
		}
	}
	@Override
	public void responseLogin(SocketChannel sc,BaseInfo baseInfo, String password) {
		System.out.println("有人登录了");
		System.out.println("得到登录者id："+baseInfo.getSenderId());
//		查询数据库，确实用户合法性
		Jedis jedis = RedisUtils.getJedis();
		jedis.zadd("login:", baseInfo.getSenderId(), baseInfo.getSenderId()+"");
		hashMap.put("login:"+baseInfo.getSenderId(), sc);
		baseInfo.setAccepteId(baseInfo.getSenderId());
		baseInfo.setSenderId(-1l);//系统发送
		baseInfo.setContent("登录成功");
		try {
			sc.write(BaseBeanUtils.lengByteBuffer(baseInfo));
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println("登录异常");
		}
		RedisUtils.returnResource(jedis);
	}
	@Override
	public void receiveSend(BaseInfo baseInfo){
		Jedis jedis = RedisUtils.getJedis();
		jedis.lpush("message", JSON.toJSON(baseInfo).toString());
		RedisUtils.returnResource(jedis);
	}
	/**
	 * 需要作业处理
	 */
	@Override
	public void responseSend(ConcurrentHashMap<String,Object> hashMap,BaseInfo baseInfo) {
		SocketChannel sc = (SocketChannel)hashMap.get("login:"+baseInfo.getAccepteId());
		if(sc!=null){
			try {
				sc.write(BaseBeanUtils.lengByteBuffer(baseInfo));
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("登录异常");
			}
		}else{
//			保存到消息缓存区域
			Jedis jedis = RedisUtils.getJedis();
			jedis.lpush("server_buffer:"+baseInfo.getAccepteId(), JSON.toJSONString(baseInfo));
			RedisUtils.returnResource(jedis);
		}
	}
	/**
	 * 需要作业处理
	 */
	@Override
	public void responseSend(SocketChannel sc,BaseInfo baseInfo) {
		if(sc!=null){
			try {
				sc.write(BaseBeanUtils.lengByteBuffer(baseInfo));
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("登录异常");
			}
		}
	}
	@Override
	public void responseLoginOut(SocketChannel sc,BaseInfo baseInfo) {
		try {
			sc.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	/**
	 * 通过多线程，来轮询内存中，是否有待发送的消息；
	 * @author Administrator
	 *
	 */
	class ThreadResponseSend implements Runnable{
		
		Thread t;
		
		ThreadResponseSend(){
		     t = new Thread(this, "ThreadResponseSend-io");
		     t.start(); // 开始线程
		}
		@Override
		public void run() {
			Jedis jedis = RedisUtils.getJedis();
			while(true){
				String result = jedis.rpop("message");
				if(result != null&&!"".equals(result)){
					BaseInfo baseInfo = JSON.parseObject(result, BaseInfo.class);
					responseSend(hashMap,baseInfo);
				}
			}
		}
	}
	/**
	 * 过滤掉已断开的连接，此功能不支持集群，需要加上server端唯一编号，来进行区分；
	 * @author Administrator
	 *
	 */
	class ThreadFilterSocketChannel implements Runnable{
		
		Thread t;
		
		ThreadFilterSocketChannel(){
			 t = new Thread(this, "ThreadFilterSocketChannel-io");
		     t.start();
		}

		@Override
		public void run() {
			Jedis jedis = RedisUtils.getJedis();
			while(true){
//				遍历jdk内存中的登录信息
				for (Entry<String, Object> e: hashMap.entrySet()) {
					SocketChannel sc = (SocketChannel)e.getValue();
					if(sc==null||!sc.isConnected()){
						hashMap.remove(e.getKey());
//						需完善redis事物，下边的for会有脏读
//		        	    jedis.zrem("login:", e.getKey().replaceAll("login:", ""));
						System.out.println("关闭jdk用户连接");
					}
				}
//				遍历redis内存中的登录信息 每次遍历100个数据信息
				Set<String> ss = jedis.zrangeByScore("login:", -1, 1000000000, 0, 100);
				for (String string : ss) {
					SocketChannel sc = (SocketChannel)hashMap.get("login:"+string);
					if(sc==null||!sc.isConnected()){
						hashMap.remove(string);
						jedis.zrem("login:", string);
						System.out.println("关闭redis用户连接");
					}
				}
			}
		}
		
	}
	/**
	 * 检查登录的用户是否有 待接收的消息，如果有就发送出去；
	 * @author Administrator
	 *
	 */
	class ThreadServerBuffer implements Runnable{
		
	Thread t;
		
	ThreadServerBuffer(){
			 t = new Thread(this, "ThreadServerBuffer-io");
		     t.start();
		}
		@Override
		public void run() {
			Jedis jedis = RedisUtils.getJedis();
			while(true){
//				遍历jdk内存中的登录信息 
				for (Entry<String, Object> e: hashMap.entrySet()) {
					String s = e.getKey().replace("login:","");
					String rr = null;
					while((rr =jedis.rpop("server_buffer:"+s))!=null){
						System.out.println("发送缓存区消息："+rr);
						responseSend((SocketChannel)e.getValue(), JSON.parseObject(rr, BaseInfo.class));
					}
				}
			}
		}
		
	}
	@Override
	public void heartBeat(SocketChannel sc, BaseInfo baseInfo) {
		baseInfo.setAccepteId(baseInfo.getSenderId());
		baseInfo.setSenderId(-1l);//系统发送
		baseInfo.setContent("心跳包回复");
	}
}
